/*
 * Copyright 2013 Eduard Sedov.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.es.util;

import java.io.IOException;
import java.nio.ByteBuffer;

/**
 *
 * @author Eduard Sedov
 */
public class LinearHashing {

    public final static int BUCKET_SIZE = 4 * 1024;
    public final static int MAX_ENTRY_SIZE = BUCKET_SIZE - Bucket.HEADER_SIZE - Bucket.ARRAY_CELL_SIZE - Bucket.KEY_LENGTH_SIZE - Bucket.TAIL_SIZE;
    private int m = 1;
    private int l = 0;
    private int p = 0;
    private BucketStorage storage;
    private boolean rehashing = false;

    public LinearHashing(String dir) throws IOException {
        this(new BucketStorage(dir));
    }

    public LinearHashing(BucketStorage storage) throws IOException {
        this.storage = storage;
        storage.getBucket(0, true);
    }

    public void close() throws IOException {
        storage.close();
    }

    public static void main(String[] args) throws IOException {
        int count = 1000;
        LinearHashing linearHashing = new LinearHashing(".\\linear");
        linearHashing.split();
        linearHashing.merge();


        long s = System.nanoTime();
        for (int i = 0; i < count; i++) {
            linearHashing.split();
        }
        System.out.println(System.nanoTime() - s);

        System.out.println("--------------------------------");
        s = System.nanoTime();
        for (int i = 0; i < count; i++) {
            linearHashing.merge();
        }
        System.out.println(System.nanoTime() - s);

    }

    public void put(byte[] key, byte[] value) {
        int bucketIndex = bucketIndex(hash(key));
        Bucket bucket = getBucket(bucketIndex);
        put(ByteBuffer.wrap(key), ByteBuffer.wrap(value), bucket);
    }

    private Bucket put(ByteBuffer key, ByteBuffer value, Bucket bucket) {
        boolean overflow = false;
        while (!bucket.hasSpace(key, value)) {
            int overflowBucketIndex = bucket.nextBucket();
            overflow = overflowBucketIndex == 0;
            bucket = storage.getOverflowBucket(bucket);
        }
        bucket.put(key, value);
        if (overflow) {
            split();
        }
        return bucket;
    }

    public byte[] get(byte[] key) {
        int bucketIndex = bucketIndex(hash(key));
        Bucket bucket = getBucket(bucketIndex);
        int index = bucket.search(key);
        while (index < 0) {
            int overflowBucketIndex = bucket.nextBucket();
            if (overflowBucketIndex == 0) {
                return null;
            } else {
                bucket = getOverflowBucket(bucket);
                index = bucket.search(key);
            }
        }
        ByteBuffer valueBuffer = bucket.getValue(index);
        byte[] value = new byte[valueBuffer.remaining()];
        valueBuffer.get(value);
        return value;
    }

    public void remove(byte[] key) {
        int bucketIndex = bucketIndex(hash(key));
        Bucket bucket = getBucket(bucketIndex);
        int index = bucket.search(key);
        while (index < 0) {
            int overflowBucketIndex = bucket.nextBucket();
            if (overflowBucketIndex == 0) {
                return;
            }
            bucket = getOverflowBucket(bucket);
            index = bucket.search(key);
        }
        bucket.removeItem(index);
        if (bucket.count() == 0) {
            merge();
        }
    }

    private int hash(ByteBuffer array) {
        if (array == null) {
            return 0;
        }
        int result = 1;
        int limit = array.limit();
        for (int i = array.position(); i < limit; i++) {
            result = 31 * result + (int) array.get(i);
        }
        return result;
    }

    private int hash(byte[] array) {
        if (array == null) {
            return 0;
        }
        int result = 1;
        for (byte element : array) {
            result = 31 * result + element;
        }
        return result;
    }

    private void split() {
        if (!rehashing) {
            rehashing = true;
            try {
                int splitBucket = p;
                int newSplitBucket = p + m * pow(l);

                p++;
                if (p == (m * pow(l))) {
                    l++;
                    p = 0;
                }

                // rehash
                rehashSplit(splitBucket, newSplitBucket);
            } finally {
                rehashing = false;
            }
        }
    }

    /**
     * Distribute items between the old and new split buckets.
     *
     * @param bucketIndex index of the old split bucket
     * @param newBucketIndex index of the new split bucket
     */
    private void rehashSplit(int bucketIndex, int newBucketIndex) {
        Bucket sourceBucket = getBucket(bucketIndex);
        Bucket targetBucket = storage.getBucket(newBucketIndex, true);

        while (true) {
            // redistribute items between
            // current source and current target buckets
            int count = sourceBucket.count();
            for (int i = count - 1; i >= 0; i--) {
                ByteBuffer key = sourceBucket.getKey(i);
                if (bucketIndex(hash(key)) == newBucketIndex) {
                    ByteBuffer value = sourceBucket.getValue(i);
                    targetBucket = put(key, value, targetBucket);
                    sourceBucket.removeItem(i);
                }
            }
            if (sourceBucket.bucketIndex() != bucketIndex) {
                // move remaining items from the current
                // to the previous source bucket
                Bucket previousBucket = getBucket(bucketIndex);
                while (sourceBucket.count() > 0 && sourceBucket.bucketIndex() != previousBucket.bucketIndex()) {
                    moveItems(sourceBucket, previousBucket);
                    if (previousBucket.nextBucket() == 0) {
                        break;
                    } else {
                        previousBucket = getOverflowBucket(previousBucket);
                    }
                }
            }
            if (sourceBucket.nextBucket() == 0) {
                // we are done if no more overflow buckets exist
                break;
            }
            sourceBucket = getOverflowBucket(sourceBucket);
        }

        // release empty overflow buckets
        Bucket bucket = getBucket(bucketIndex);
        if (bucket.nextBucket() > 0) {
            do {
                bucket = getOverflowBucket(bucket);
            } while (bucket.count() > 0 && bucket.nextBucket() > 0);

            if (bucket.count() == 0) {
                storage.releaseOverflowBucket(bucket);
            }
        }
    }

    /**
     * Moves items from source to target bucket until it is full.
     *
     * @param source
     * @param target
     */
    private void moveItems(Bucket source, Bucket target) {
        int count = source.count();
        for (int i = count - 1; i >= 0; i--) {
            ByteBuffer key = source.getKey(i);
            ByteBuffer value = source.getValue(i);
            if (target.hasSpace(key, value)) {
                source.removeItem(i);
                target.put(key, value);
            }
        }
    }

    private void merge() {
        rehashing = true;
        try {
            p--;
            if (p < 0 && l > 0) {
                l--;
                p = m * pow(l) - 1;
            }
            if (p < 0) {
                p = 0;
            }
            if (l > 0) {

                int mergeBucket = p;
                int oldMergeBucket = p + m * pow(l);

                // rehash
                rehashMerge(mergeBucket, oldMergeBucket);
            }
        } finally {
            rehashing = false;
        }
    }

    private void rehashMerge(int bucketIndex, int oldBucketIndex) {
        Bucket sourceBucket = storage.getBucket(oldBucketIndex, false);
        Bucket targetBucket = getBucket(bucketIndex);

        while (true) {
            // move all items from source to target bucket
            while (sourceBucket.count() > 0) {
                ByteBuffer key = sourceBucket.getKey(sourceBucket.count() - 1);
                ByteBuffer value = sourceBucket.getValue(sourceBucket.count() - 1);
                targetBucket = put(key, value, targetBucket);
                sourceBucket.removeItem(sourceBucket.count() - 1);
            }
            if (sourceBucket.nextBucket() > 0) {
                sourceBucket = getOverflowBucket(sourceBucket);
            } else {
                break;
            }
        }

        storage.releaseBucket(storage.getBucket(oldBucketIndex, false));
    }

    private int pow(int pow) {
        return 1 << pow;
    }

    private Bucket getBucket(int index) {
        return storage.getBucket(index, index >= p + m * pow(l));
    }

    private Bucket getOverflowBucket(Bucket bucket) {
        return storage.getOverflowBucket(bucket);
    }

    private int bucketIndex(int hash) {
        int index = hash % (m * pow(l));
        if (index < p) {
            return hash % (m * pow(l + 1));
        }
        return index;
    }

    private void print() {
        int maxBucketIndex = p + m * pow(l) - 1;
        for (int bucketIndex = 0; bucketIndex < maxBucketIndex; bucketIndex++) {
            System.out.println(" ------------- " + bucketIndex + " ------------ ");
            Bucket bucket = getBucket(bucketIndex);
            print(bucket);
        }
    }

    private void print(Bucket bucket) {
        for (int i = 0; i < bucket.count(); i++) {
            ByteBuffer key = bucket.getKey(i);
            ByteBuffer value = bucket.getValue(i);
            System.out.println(i + "\t"
                    + new String(key.array(), key.arrayOffset(), key.remaining()) + "\t"
                    + new String(value.array(), value.arrayOffset(), value.remaining()));
        }
        if (bucket.nextBucket() > 0) {
            bucket = getOverflowBucket(bucket);
            print(bucket);
        }
    }
}
